package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.common.GmallConfig;
import com.atguigu.gmall.realtime.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Collection;
import java.util.Set;

/**
 * Author: Felix
 * Date: 2021/11/6
 * Desc: 将维度侧输出流中的数据写到phoenix表中
 */
public class DimSink extends RichSinkFunction<JSONObject> {
    private Connection conn;
    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        //建立连接
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        PreparedStatement ps = null;
        //获取插入的维度表表名
        String sinkTable = jsonObj.getString("sink_table");
        //获取json中的data属性
        JSONObject dataJsonObj = jsonObj.getJSONObject("data");
        try{

            //拼接插入SQL   upsert into 库.表 values();
            String upsertSql = genUpsertSql(sinkTable,dataJsonObj);
            System.out.println("执行的upsert语句为:" + upsertSql);
            //创建数据库操作对象
            ps = conn.prepareStatement(upsertSql);
            //执行SQL语句
            ps.executeUpdate();
            //注意：使用jdbc向phoenix中插入数据，需要手动提交事务
            conn.commit();
        }catch (Exception e){
            e.printStackTrace();
            throw new RuntimeException("向维度表中插入数据失败~");
        }finally {
            //释放资源
            if(ps != null){
                ps.close();
            }
        }

        //如果维度表的数据被修改 ，那么清空Redis中缓存的维度数据
        if("update".equals(jsonObj.getString("type"))){
            DimUtil.deleteCached(sinkTable,dataJsonObj.getString("id"));
        }

    }

    //生成插入的sql
    private String genUpsertSql(String tableName, JSONObject dataJsonObj) {
        Set<String> keys = dataJsonObj.keySet();
        Collection<Object> values = dataJsonObj.values();
        String upsertSql = "upsert into "+GmallConfig.HBASE_SCHEMA+"."
            +tableName+" ("+ StringUtils.join(keys,",") +") " +
            " values('"+StringUtils.join(values,"','")+"')";
        return upsertSql;
    }
}
